package storm.online.topology;

import storm.online.bolt.AreaOnlineBolt;
import storm.online.bolt.GameOnlineBolt;
import storm.online.bolt.LogBehaviorBolt;
import storm.online.bolt.LogBehaviorPrinterBolt;
import storm.online.bolt.LogInfoParseBolt;
import storm.online.bolt.LogInfoPrinterBolt;
import storm.online.spout.LoginMessageSpout;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class OnlineNumTopology {

	public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        
        builder.setSpout("spout", new LoginMessageSpout(), 1);
//        builder.setBolt("debug", new LogInfoPrinterBolt(), 1).shuffleGrouping("spout");
        builder.setBolt("parser", new LogInfoParseBolt(), 1).shuffleGrouping("spout");
        builder.setBolt("behavior", new LogBehaviorBolt(), 3).fieldsGrouping("parser", new Fields("game_id", "area_id"));
//        builder.setBolt("debug", new LogBehaviorPrinterBolt(), 1).shuffleGrouping("behavior");
        builder.setBolt("areaOlnum", new AreaOnlineBolt(), 2).fieldsGrouping("behavior", new Fields("game_id", "area_id"));
        builder.setBolt("gameOlnum", new GameOnlineBolt(), 2).fieldsGrouping("behavior", new Fields("game_id"));
        
        Config conf = new Config();
        conf.setDebug(true);
        
        if(args!=null && args.length > 0) {
            conf.setNumWorkers(3);
            
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {        
            conf.setMaxTaskParallelism(3);

            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("online-number", conf, builder.createTopology());
        
            Thread.sleep(30000);

            cluster.shutdown();
        }
	}
	
}
